package org.example.queueadmin.task.runner;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.example.queueadmin.dlq.DeadLetterService;
import org.example.queueadmin.log.MessageLogService;
import org.example.queueadmin.task.TaskRunner;
import org.example.queueadmin.task.TransferTask;
import org.example.queueadmin.task.TransferTaskRepository;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.time.Instant;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@RequiredArgsConstructor
public class RabbitToKafkaRunner implements TaskRunner {
  private final TransferTask task;
  private final DeadLetterService dlq;
  private final TransferTaskRepository repository;
  private final MessageLogService messageLogService;
  private SimpleMessageListenerContainer container;
  private KafkaProducer<byte[], byte[]> producer;
  private final AtomicBoolean running = new AtomicBoolean(false);

  @Override
  public synchronized void start() {
    if (running.get()) return;
    running.set(true);

    // Rabbit 消费
    CachingConnectionFactory cf = new CachingConnectionFactory(task.getRabbitSourceHost(), task.getRabbitSourcePort());
    cf.setUsername(task.getRabbitSourceUsername());
    cf.setPassword(task.getRabbitSourcePassword());
    cf.setVirtualHost(task.getRabbitSourceVhost());

    container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(cf);
    container.setQueueNames(task.getRabbitSourceQueue());
    container.setAcknowledgeMode(org.springframework.amqp.core.AcknowledgeMode.MANUAL);
    if (task.getRabbitSourcePrefetch() != null) {
      container.setPrefetchCount(task.getRabbitSourcePrefetch());
    }

    // 应用并发度配置
    if (task.getConcurrency() != null && task.getConcurrency() > 0) {
      container.setConcurrentConsumers(task.getConcurrency());
      container.setMaxConcurrentConsumers(task.getConcurrency());
    }

    container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
      String source = task.getRabbitSourceQueue();
      String target = task.getKafkaTargetTopic();
      byte[] body = message.getBody();

      try {
        ensureKafkaProducer();
        ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(task.getKafkaTargetTopic(), body);
        producer.send(record, (metadata, exception) -> {
          if (exception != null) {
            log.error("Rabbit->Kafka 发送失败: {}", exception.getMessage(), exception);
            dlq.push(task.getId(), body, exception.getMessage());

            // 记录失败日志
            messageLogService.logMessage(task.getId(), "Rabbit→Kafka", source, target, body, false, exception.getMessage());
            updateFailedCount(exception.getMessage());
          } else {
            // 记录成功日志
            messageLogService.logMessage(task.getId(), "Rabbit→Kafka", source, target, body, true, null);
            updateProcessedCount();
          }
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
      } catch (Exception e) {
        log.error("Rabbit->Kafka 处理失败: {}", e.getMessage(), e);
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        dlq.push(task.getId(), message.getBody(), e.getMessage());

        // 记录失败日志
        messageLogService.logMessage(task.getId(), "Rabbit→Kafka", source, target, body, false, e.getMessage());
        updateFailedCount(e.getMessage());
      }
    });
    container.start();
  }

  private void ensureKafkaProducer() {
    if (producer != null) return;
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, task.getKafkaTargetBootstrap());
    props.put(ProducerConfig.ACKS_CONFIG, task.getKafkaTargetAcks() == null ? "all" : task.getKafkaTargetAcks());
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

    // 应用重试配置
    props.put(ProducerConfig.RETRIES_CONFIG, task.getMaxRetries() != null ? task.getMaxRetries() : 3);
    props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, task.getRetryBackoffMs() != null ? task.getRetryBackoffMs() : 1000);

    producer = new KafkaProducer<>(props);
  }

  @Override
  public synchronized void stop() {
    running.set(false);

    if (container != null) {
      try {
        log.info("Stopping RabbitMQ container for task: {}", task.getName());
        container.stop();
        container = null;
        log.info("RabbitMQ container stopped for task: {}", task.getName());
      } catch (Exception e) {
        log.warn("Error stopping RabbitMQ container for task {}: {}", task.getName(), e.getMessage());
        container = null;
      }
    }

    if (producer != null) {
      try {
        log.info("Closing Kafka producer for task: {}", task.getName());
        producer.close();
        producer = null;
        log.info("Kafka producer closed for task: {}", task.getName());
      } catch (Exception e) {
        log.warn("Error closing Kafka producer for task {}: {}", task.getName(), e.getMessage());
        producer = null;
      }
    }
  }

  @Override
  public synchronized void restart() { stop(); start(); }

  @Override
  public String status() { return container != null && running.get() ? "RUNNING" : "STOPPED"; }

  private void updateProcessedCount() {
    try {
      TransferTask t = repository.findById(task.getId()).orElse(null);
      if (t != null) {
        t.setProcessed((t.getProcessed() == null ? 0 : t.getProcessed()) + 1);
        repository.save(t);
      }
    } catch (Exception e) {
      log.warn("Failed to update processed count: {}", e.getMessage());
    }
  }

  private void updateFailedCount(String errorMsg) {
    try {
      TransferTask t = repository.findById(task.getId()).orElse(null);
      if (t != null) {
        t.setFailed((t.getFailed() == null ? 0 : t.getFailed()) + 1);
        t.setLastErrorAt(Instant.now());
        t.setLastErrorMsg(errorMsg);
        repository.save(t);
      }
    } catch (Exception e) {
      log.warn("Failed to update failed count: {}", e.getMessage());
    }
  }
}

